package com.shujia.spark.core

import com.shujia.spark.util.HDFSUtil
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo1WordCount {
  def main(args: Array[String]): Unit = {

    /**
      * 创建spark环境
      *
      */

    //配置对象
    val conf = new SparkConf()

    //设置任务名
    conf.setAppName("wordcount")

    //指定spark代码运行方式，local:本地测试
    conf.setMaster("local")

    //spark 上下文对象，用于读取数据
    val sc = new SparkContext(conf)


    /**
      * 大数据处理引擎步骤
      * 1、读取数据
      * 2、处理数据
      * 3、保存结果
      *
      */

    /**
      *
      * RDD: 弹性的分布式数据集
      */


    //1、读取数据
    //暂时可以将rdd当成scala中的list使用
    //如果是提交到集群运行，会自动切换为读取hdfs数据
    val linesRDD: RDD[String] = sc.textFile("data/words")


    //2、将数据展开
    val wordsRDD: RDD[String] = linesRDD.flatMap(line => line.split(","))


    //3、按照单词进行分组
    //分组之后key是一个单词
    //value是一个迭代器
    //迭代器中的数据不会全部一下加载到内存中
    val groupByRDD: RDD[(String, Iterable[String])] = wordsRDD.groupBy(word => word)


    //4、统计单词的数量
    val wordCountRDD: RDD[String] = groupByRDD.map {
      case (word: String, iter: Iterable[String]) =>
        //计算单词的数量
        val count: Int = iter.size

        //返回数据
        s"$word\t$count"
    }


    /**
      * 删除输出目录， 使用hdfs api类删除输出目录
      *
      */
    HDFSUtil.deletePath("data/wordcount")


    //5、保存数据
    //如果是再集群中运行，会保存到hdfs
    wordCountRDD.saveAsTextFile("data/wordcount")

  }

}
